package profile.tagmerge

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.SparkSession
import profile.tagcombine2.TagCombineBean2
import profile.utils.{TagConfigUtil, TagMergeUtil}

/**
  * Created by hunter.coder 涛哥  
  * 2019/4/24 16:40
  * 交流qq:657270652
  * Version: 1.0
  * 更多学习资料：https://blog.csdn.net/coderblack/
  * Description:
  **/
object TagMerge {

  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.WARN)

    val spark = SparkSession.builder().master("local").appName("标签最终合并").getOrCreate()
    import spark.implicits._

    val oldTag = spark.read.parquet("G:\\testdata\\usertags\\out-combined-tags\\T-1日")
    val newTag = spark.read.parquet("G:\\testdata\\usertags\\out-combined-tags\\T日")

    oldTag.show(10, false)
    newTag.show(10, false)

    oldTag.createTempView("old")
    newTag.createTempView("new")

    val joinTag = spark.sql(
      """
        |
        |select
        |if(o.gid is null,n.gid,o.gid) as gid
        |o.idsTags as o_idsTags,
        |o.deviceTags as o_deviceTags,
        |o.goodsKwTags as o_goodsKwTags,
        |o.dspKwTags as o_dspKwTags,
        |
        |n.idsTags as n_idsTags,
        |n.deviceTags as n_deviceTags,
        |n.goodsKwTags as n_goodsKwTags,
        |n.dspKwTags as n_dspKwTags
        |
        |from old o
        |full outer join
        |new n on o.gid = n.gid
        |
      """.stripMargin)

    joinTag.printSchema()
    joinTag.show(10, false)


    val mergedTags = joinTag.rdd.map(row => {
      /**
        * 合并idsTags标签模块
        */
      val gid = row.getAs[String]("gid")
      val oidsTags = row.getAs[Map[String, Map[String, Double]]]("o_idsTags")
      val nidsTags = row.getAs[Map[String, Map[String, Double]]]("n_idsTags")
      val mIdsTags: Map[String, Map[String, Double]] = TagMergeUtil.mergeReductionTagMap(
        if (oidsTags != null) oidsTags else Map.empty[String, Map[String, Double]],
        if (nidsTags != null) nidsTags else Map.empty[String, Map[String, Double]],
        TagConfigUtil.getTagReductionFactor("idsTags"))

      /**
        * 合并deviceTags标签模块
        */


      /**
        * 合并goodsKwdsTags标签模块
        */


      /**
        * 合并dspKwTags标签模块
        */


      /**
        * 合并goodsKwdsTags标签模块
        */


      /**
        * 合并locTags标签模块
        */


      /**
        * 合并appTags标签模块
        */


      /**
        * 合并adTags标签模块
        */


      /**
        * 合并doitBhvTags标签模块
        */


      /**
        * 合并doitBuyTags标签模块
        */


      TagCombineBean2(gid,mIdsTags,null,null,null,null,null,null,null,null)
    })
        .toDS()
        .write.parquet("g:/testdata/usertag/mergedTags-T")


    spark.close()
  }

}
